package com.github.jiangxch.rpc.transport.netty;

import com.github.jiangxch.rpc.serialization.Serializer;
import com.github.jiangxch.rpc.transport.AbstractEndpoint;
import com.github.jiangxch.rpc.transport.Channel;
import com.github.jiangxch.rpc.transport.ChannelHandler;
import com.github.jiangxch.rpc.transport.Client;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

/**
 * @author: jiangxch
 * @date: 2021/2/20 11:11
 */
@Slf4j
public class NettyClient extends AbstractEndpoint implements Client {
    private ChannelHandler channelHandler;
    private InetSocketAddress connectAddress;
    private ChannelFuture channelFuture;
    private volatile boolean isClosed = true;
    private NioEventLoopGroup group;
    
    public NettyClient(InetSocketAddress connectAddress, ChannelHandler channelHandler, Serializer serializer) {
        super(serializer);
        this.connectAddress = connectAddress;
        this.channelHandler = channelHandler;
        
    }
    
    private void initNettyClient(){
        group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));
                            pipeline.addLast(new NettyChannelInboundHandler(channelHandler,NettyClient.this));
                        }
                    });
            channelFuture = b.connect(connectAddress.getHostString(),connectAddress.getPort());
            channelFuture.awaitUninterruptibly(3000, TimeUnit.MILLISECONDS);
            log.info("rpc client socket start success");
        } catch (Exception e) {
            log.info("rpc client socket start exception",e);
        }
    }
    
    @Override
    public InetSocketAddress getLocalSocketAddress() {
        return null;
    }
    
    @Override
    public void start() {
        this.initNettyClient();
        this.isClosed = false;
    }
    
    @Override
    public void close() {
        // 关闭阻塞
        channelFuture.channel().closeFuture().addListener(new GenericFutureListener<Future<? super Void>>() {
            @Override
            public void operationComplete(Future<? super Void> future) throws Exception {
                if (future.isSuccess()) {
                    isClosed = true;
                    group.shutdownGracefully();
                }
            }
        });
    }
    
    @Override
    public boolean isClosed() {
        return isClosed;
    }
    
    @Override
    public Channel getChannel() {
        return NettyChannel.wrap(channelFuture.channel(),serializer);
    }
}
